package com.fjeld.middleware.activemq;

import javax.jms.*;

import com.fjeld.middleware.constants.PropertiesManager;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;

import java.io.Serializable;

import com.fjeld.server.store.reader.XMLKit;

/**
 * A topic publisher, publishing serialized objects on the given topic.
 */
public class TopicPublisher implements MessageListener
{
    private static final Logger LOGGER = Logger.getLogger(TopicPublisher.class);

    private Session session;
    private Connection connection;
    private MessageProducer publisher;
    protected Topic topic;
    protected String topicName, publisherName = "Default", logPrefix, hostName, hostPort, brokerURL;

    public TopicPublisher(String stringTopic)
    {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(BrokerEnum.BrokerURL.getValue());
        try
        {
            connection = factory.createConnection();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic(stringTopic);
            topicName = topic.getTopicName();
            logPrefix = "[" + publisherName + " on " + topicName + "] ";

            publisher = session.createProducer(topic);
            publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            connection.start();
            LOGGER.info("[" + logPrefix + "] TopicPublisher created");
        } catch (JMSException e)
        {
            LOGGER.error(logPrefix + ": Problem when creating TopicPublisher: ", e);
        }
    }

    public TopicPublisher(String topicString, String publisherName)
    {
        // Should be in the form of server.service (e.g. ContentStoreServer.HeartbeatPublisher
        this.publisherName = publisherName;

        hostName = PropertiesManager.getProperties().getProperty(publisherName + ".HostName", "localhost");
        hostPort = PropertiesManager.getProperties().getProperty(publisherName + ".HostPort", "6161");
        brokerURL = "tcp://" + hostName + ":" + hostPort;
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);

        try
        {
            connection = factory.createConnection();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic(topicString);
            topicName = topic.getTopicName();
            logPrefix = "[" + publisherName + " on " + topicName + "] ";

            publisher = session.createProducer(topic);
            publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            LOGGER.info(logPrefix + "TopicPublisher has been created");
        } catch (JMSException e)
        {
            LOGGER.error(logPrefix + ": Problem when creating TopicPublisher: ", e);
        }
    }

    public void startConnection()
    {
        try
        {
            LOGGER.debug(logPrefix + "Starting publisher transport...");
            connection.start();
            LOGGER.debug(logPrefix + "Started publisher transport.");
        } catch (JMSException e)
        {
            LOGGER.fatal(logPrefix + "Could not start publisher.", e);
            System.exit(1);
        }


    }

    /**
     * Server request shutdown of all clients
     *
     * @throws JMSException
     */
    public void shutdown()
    {
        try
        {
            LOGGER.debug(logPrefix + "Shutting down...");
            publisher.send(session.createTextMessage("SHUTDOWN"));
            connection.stop();
            connection.close();
            LOGGER.debug(logPrefix + "Shut down.");
        } catch (JMSException e)
        {
            LOGGER.error(logPrefix + "Shutdown failed!");
            e.printStackTrace();
        }
    }

    /**
     * Publish a message with the object
     *
     * @param object
     * @throws Exception
     */
    public void publish(Serializable object) throws Exception
    {
        ObjectMessage msg = session.createObjectMessage(object);
        publisher.send(msg);
        //LOGGER.debug(logPrefix +"Sent message:\n" + XMLKit.xmlObjToString(object));
    }

    public void onMessage(Message message)
    {
        //LOGGER.debug(logPrefix +"Received: " + message.toString());
    }

}
